/**
 * Software License, Version 1.0
 * 
 * Copyright 2008 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgment:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgment may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or Granules, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called Granules, 
 * nor may Indiana University or Community Grids Lab or Granules appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 * GENERATED USING SOFTWARE.
 */
package cgl.granules.samples.matrix;

import java.io.IOException;
import java.util.Properties;

import cgl.granules.dataset.Dataset;
import cgl.granules.dataset.DatasetCollection;
import cgl.granules.dataset.DatasetException;
import cgl.granules.dataset.StreamingAccess;
import cgl.granules.exception.CommunicationsException;
import cgl.granules.exception.MarshallingException;
import cgl.granules.operation.MapReduceBase;
import cgl.granules.operation.MapReduceException;
import cgl.granules.results.ResultPayload;
import cgl.granules.results.Results;
import cgl.granules.scheduler.StreamingService;
import cgl.narada.event.NBEvent;
import cgl.narada.service.ServiceException;
import cgl.narada.service.client.EventProducer;

/**
 * An example demonstrating the <code>Map</code> role within a Parallel-sort
 * operation.
 * 
 * @author Shrideep Pallickara
 * 
 */
public class MatrixMap extends MapReduceBase {

	private int[] row = null;
	private int[] column = null;
	private int[] computedColumn = null;

	// private String workUnitStream = "FinishedWorkUnit/MatrixElement";
	private String computedColumnStream = "Computed/Column/";
	private String rowStreamIndicator = "MatrixStreams/Row/";
	private String columnStreamIndicator = "MatrixStreams/Column";
	private ResultPayload resultPayload;
	private boolean debug = false;

	private boolean initializedDatastructures = false;
	private int columnID = 0;
	private int numOfRowsToProcess = 1000;
	private int computations = 0;
	private MatrixSharedMemory sharedMemory;
	private EventProducer producer;
	private StreamingService streamingService;

	public MatrixMap() {
		setDomain("Generic");
		setVersionInformation(200);
		setOperationIdentifier("ABCDEFGHIJKLMN");
		resultPayload = ResultPayload.getInstance();
		sharedMemory = MatrixSharedMemory.getInstance();
		streamingService = StreamingService.getInstance();
	}

	private void initializeDatastructures() {
		initializedDatastructures = true;

		long memUtilization = getMemoryUtilization();
		System.out.println("Matrix Map: Memory utilization is: "
				+ memUtilization + "MB ");

		Properties props = getProcessingDirectives();

		String property = "matrix column";
		if (props.containsKey(property)) {
			columnID = Integer.parseInt(props.getProperty(property));
		}

		property = "entries";
		if (props.containsKey(property)) {
			numOfRowsToProcess = Integer.parseInt(props.getProperty(property));
		}

		computedColumn = new int[numOfRowsToProcess];
	}

	public void execute() {
		if (!initializedDatastructures) {
			initializeDatastructures();
		}

		int workUnits = 0;
		if (!hasDatasetCollection()) {
			System.out
					.println("No dataset has been initialized. Returning ...");
			return;
		}

		String datasetIdentifier = "DatasetID";
		DatasetCollection datasetCollection = getDatasetCollection();
		StreamingAccess access = null;
		try {
			Dataset dataset = datasetCollection.getDataset(datasetIdentifier);

			if (dataset.getDatasetType() == Dataset.STREAMS) {
				access = (StreamingAccess) dataset;
			} else {
				System.out.println("Incorrect datatset: "
						+ dataset.getDatasetType()
						+ " initialized. Returning ...");
				return;
			}

			if (access != null) {

				while (dataset.isDataAvailable()) {
					NBEvent nbEvent = access.getStreamingData();
					processStreamPacket(nbEvent);
					workUnits++;
				}

				if (computations == numOfRowsToProcess) {
					row = null;
					column = null;
					sendComputedColumn();
					access.close();
					setTeminationConditionReached();
				}
			}

		} catch (DatasetException e) {
			e.printStackTrace();
		} catch (IOException e) {
			System.out.println("Problems converting bytes to longs");
			e.printStackTrace();
		} catch (MarshallingException e) {
			System.out.println("Problems Marshalling: " + e.getMessage());
			e.printStackTrace();
		} catch (MapReduceException e) {
			System.out.println("Problems writing Results: " + e.getMessage());
			e.printStackTrace();
		}
		// System.out.println("Finished :" + workerType + ", which did ("
		// + workUnits + ") units of work on the data.");
		catch (ServiceException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (CommunicationsException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}

	private void calculateElementInProductMatrix(int rowId)
			throws MarshallingException, MapReduceException {
		int size = row.length;
		if (size != column.length) {
			String report = "Mismatch in the size of the arrays to be used for"
					+ " computing the element in the product matrix. Size of the "
					+ "row=" + row.length + ", size of column=" + column.length;
			// System.out.println(report);
		}

		int value = 0;
		for (int i = 0; i < size; i++) {
			value += (row[i] * column[i]);
		}

		computedColumn[rowId] = value;

		if (debug) {
			String report = "Computed [" + rowId + "," + columnID + "]  -> "
					+ getMemoryUtilization() + " MB";
			System.out.println(report);
		}

		computations++;

		/*
		 * ResultPayload resultPayload = ResultPayload.getInstance(); byte[]
		 * resultPayloadBytes = resultPayload.getBytes(value);
		 * 
		 * Results results = createResults(true, true);
		 * 
		 * results.setResultPayload(resultPayloadBytes);
		 * writeResults(workUnitStream, results);
		 */

	}

	private void processStreamPacket(NBEvent nbEvent) throws IOException,
			MarshallingException, MapReduceException {
		String synopsis = (String) nbEvent.getContentSynopsis();

		if (synopsis.startsWith(rowStreamIndicator)) {
			byte[] payload = nbEvent.getContentPayload();
			// row = resultPayload.getIntArray(payload);
			row = sharedMemory.getRow(synopsis, nbEvent);
			String row = synopsis.substring(rowStreamIndicator.length());
			int rowId = Integer.parseInt(row);
			if (debug) {
				System.out.println("Row {" + row + "}");
			}

			// if (row != null && column != null) {
			calculateElementInProductMatrix(rowId);
			sharedMemory.returnRow(synopsis);
			// }
			// row = null;
		}

		if (synopsis.startsWith(columnStreamIndicator)) {
			byte[] payload = nbEvent.getContentPayload();
			column = resultPayload.getIntArray(payload);
		}
	}

	private long getMemoryUtilization() {
		Runtime runtime = Runtime.getRuntime();
		long totalMemory = runtime.totalMemory();
		long freeMemory = runtime.freeMemory();

		long memUtilization = totalMemory - freeMemory;
		return (memUtilization / (1024 * 1024));
	}

	private void sendComputedColumn() throws MarshallingException,
			MapReduceException, ServiceException, CommunicationsException {
		ResultPayload resultPayload = ResultPayload.getInstance();
		boolean completeResult = false; /* We are sending only a column */
		boolean processingComplete = true; /* This map has finished */
		Results results = createResults(completeResult, processingComplete);
		results.setResultPayload(resultPayload.getBytes(computedColumn));

		System.out.println("Writing computed column [" + columnID + "]");
		String columnStream = computedColumnStream + columnID;
		writeResults(StreamingAccess.HIERARCHICAL_SYNOPSIS, columnStream,
				results);

		/**
		 * if (producer == null) { producer = streamingService.createProducer();
		 * }
		 * 
		 * NBEvent nbEvent;
		 * 
		 * nbEvent = producer.generateEvent(
		 * TemplateProfileAndSynopsisTypes.HIERARCHICAL, columnStream,
		 * resultPayload.getBytes(computedColumn));
		 * producer.publishEvent(nbEvent);
		 */
	}
}
